/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.protocol;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.StringTokenizer;

import cgl.narada.matching.tagvalue.MEvent;
import cgl.narada.matching.tagvalue.Predicate;
import cgl.narada.node.ClientNode;
import cgl.narada.transport.TransportException;
import cgl.narada.transport.TransportHandler;
import cgl.narada.transport.TransportHandlerImpl;
import cgl.narada.util.ByteUtilities;
import cgl.narada.util.MessageQueue;

/**
 * This class serves to provide the client with an intermediary between the
 * application layers and the transport layers. The purpose of this class is
 * also to provide for
 * <ul>
 * <li>The generation of connection messages.
 * <li>Creation of events - publishing and subscribing to them.
 * <li>Support the <i>induced roam </i> concept.
 * </ul>
 * 
 * @author Shrideep Pallickara $Revision$ $Date$
 */

public class ClientProtocolHandler extends Thread implements
    ProtocolDebugFlags, Handler {

    /** Destination that this node is attached to */
    private Destinations serverDestination;

    /** Destination assigned to client within this destination */
    private int destinations;

    /** Deals with the transport related aspects of communication */
    private TransportHandler transportHandler;

    private byte[] data;

    /** An individual dataUnit within the queue of received messages */
    private DataReceived dataReceived;

    private int tracker = 0;

    private boolean gatherStats = false;

    int latencyTrack = 0;

    short[] latencyArray;

    /**
     * This measurement requires a very exact understanding of the number of
     * messages that are delivered by the subscribing client. Since we have
     * control on the generation of events (the number/content) we can compute
     * system throughputs
     */
    long throughputStartTime;

    long throughputEndTime;

    long minLatency = 200000;

    long maxLatency = 0;

    int packetNumber = 0;

    private ClientNode clientNode;

    private boolean loopThroughMessageSizes = false;

    private int[] messageSizes;

    private int messageSizeTracker;

    private int numOfMessagesPerLoopIteration;

    private int[] publishRates;

    private int publishRateTracker;

    private int[] matchRates;

    private int matchRateTracker;

    private String reportFileName;

    private int publishRate;

    private int testCases = 0;

    /**
     * Constructor for the Client Protocol Handler. This initializes the
     * transport handler interfaces and the message queues needed for
     * communication.
     */
    public ClientProtocolHandler() {
        transportHandler = new TransportHandlerImpl(this);
    }

    public void setClientNode(ClientNode clientNode) {
        this.clientNode = clientNode;
    }

    /** This method indicates that the node in question is a client */
    public boolean isClient() {
        return true;
    }

    /**
     * This method is responsible for dealing with packets of messages that are
     * received over a link, based on the protocol headers that are contained
     * within the packets. The DataReceived object is the encapsulation of the
     * data that is received by one of the ReceiverThreads within the transport
     * layer. This handler is specifc to dealing with clients
     * 
     * @param _dataReceived
     *            The data that is received by one of the ReceiverThreads
     */
    public void handleData(byte[] data, String sender) {
        if (data.length == 0) {
            System.out
                .println("ClientProtocolHandler:handleDate() -> length == 0");
            return;
        }
        int dataInfo = (int) data[0];
        // System.out.println("Received -> protocol Header " + dataInfo +
        // " from " + sender);
        packetNumber++;
        switch (dataInfo) {
        case (int) Protocol.CONN_SUCC:
            /** CONN_REQ */
            System.out
                .println("ClientProtocolHandler:Connection successful to -> "
                    + sender);
            break;
        case (int) Protocol.NAP_CLIENT_ADD_RESPONSE:
            processNodeAddressResponse(sender, data);
            break;

        case (int) Protocol.EVENT:
            processEventReceived(sender, data);
            break;

        case (int) Protocol.PERSISTENT_EVENT:
            processPersistentEventReceived(sender, data);
            break;

        default:
            System.out.println("Client ProtocolHandler: Data[0] ->" + dataInfo
                + " Data length = " + data.length + "Packet Num "
                + packetNumber);
            break;
        }
    }

    public void initializeLatencyArray(int size) {
        gatherStats = true;
        latencyArray = new short[size];
        latencyTrack = 0;
        minLatency = 200000;
        maxLatency = 0;
        numOfMessagesPerLoopIteration = size;
    }

    public void setLoopThroughMessageSizes(String reportFileName) {
        loopThroughMessageSizes = true;
        messageSizes = clientNode.getMessageSizes();
        messageSizeTracker = 0;
        publishRates = clientNode.getPublishRates();
        publishRateTracker = 0;
        matchRates = clientNode.getMatchRates();
        matchRateTracker = 0;
        this.reportFileName = reportFileName;
        clientNode.initializeReportFile(reportFileName + "WithMatchRate"
            + matchRates[matchRateTracker]);
        initializeLatencyArray(matchRates[matchRateTracker]);
    }

    public short[] getLatencyArray() {
        return latencyArray;
    }

    public void setPublishRate(int _pubRate) {
        publishRate = _pubRate;
    }

    public void printTestCaseDetails() {
        int maxReject = 5;
        long[] maxLatencies = new long[maxReject];
        for (int i = 0; i < maxReject; i++) {
            maxLatencies[i] = (long) 0;
        }
        int _numOfMessagesReceived = latencyArray.length;
        int numOfMessages = _numOfMessagesReceived;

        for (int i = 0; i < numOfMessages; i++) {
            for (int j = 0; j < maxReject; j++) {
                if (latencyArray[i] >= maxLatencies[maxReject - 1 - j]) {
                    for (int k = 0; k < maxReject - 1 - j; k++) {
                        maxLatencies[k] = maxLatencies[k + 1];
                    }/* (k) */
                    maxLatencies[maxReject - 1 - j] = latencyArray[i];
                    break;
                }
                /** if statement - for reorganizing the max array */
            }/* (j) */
        }/* (i) */

        double _latency = 0;

        for (int i = 0; i < _numOfMessagesReceived; i++) {
            _latency += latencyArray[i];
        }

        for (int i = 0; i < maxReject; i++) {
            _latency -= maxLatencies[i];
        }

        double _meanLatency = _latency / (_numOfMessagesReceived - maxReject);
        if (loopThroughMessageSizes)
            System.out.println("\n" + "Individual messages were of size "
                + messageSizes[messageSizeTracker]);
        System.out.println("Mean latency for " + _numOfMessagesReceived
            + " messages is " + _meanLatency + "mSecs "
            + " Min/Max Latencies -> " + minLatency + "/" + maxLatency
            + "mSecs");

        double squaredDifferenceOfMeanAndSample = 0;
        for (int i = 0; i < _numOfMessagesReceived; i++) {
            squaredDifferenceOfMeanAndSample += (Math
                .pow((latencyArray[i] - _meanLatency), 2));
        }

        for (int i = 0; i < maxReject; i++) {
            squaredDifferenceOfMeanAndSample -= (Math
                .pow((maxLatencies[i] - _meanLatency), 2));
        }

        double variance = squaredDifferenceOfMeanAndSample
            / (_numOfMessagesReceived - maxReject - 1);
        double stdDeviation = Math.sqrt(variance);
        System.out.println("Standard Deviation for this sample of messages is "
            + stdDeviation);

        double systemThroughput = ((double) _numOfMessagesReceived / (double) (throughputEndTime - throughputStartTime)) * 1000;
        System.out.println("System throughput is " + systemThroughput
            + " Messages/Sec");

        if (loopThroughMessageSizes) {
            long waitForGatheringResults;
            try {
                waitForGatheringResults = (110000 / matchRates[matchRateTracker]);
                Thread.sleep(waitForGatheringResults);
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
            String reportLine = clientNode.readPublishRate() + "   "
                + messageSizes[messageSizeTracker] + "  "
                + (float) _meanLatency + "  " + (float) stdDeviation + "   "
                + (float) systemThroughput + "  " + minLatency + "   "
                + maxLatency;
            clientNode.writeToReportFile(reportLine);

            /** The processed sample of messages */
            long[] acceptedLatencies = new long[numOfMessages - maxReject];
            boolean accept;
            int k = 0;

            for (int i = 0; i < numOfMessages; i++) {
                accept = true;
                for (int j = 0; j < maxReject; j++) {
                    if (latencyArray[i] == maxLatencies[j]) {
                        maxLatencies[j] = (long) 100000;
                        accept = false;
                        break;
                    }
                } /* end max-reject */

                if (accept) {
                    acceptedLatencies[k++] = latencyArray[i];
                }
            } /* end latency array */

            clientNode.writeLatenciesToFile(acceptedLatencies);

        }
    }

    /**
     * This method processes the packet that is received with the Protocol.EVENT
     * header. This method is reponsible for unmarshalling the event stream and
     * creating the event associated with the stream.
     * 
     * @param sender
     *            The server from which we received this event.
     * @param receivedEventBytes
     *            The byte stream representation of the event, including the
     *            protocol headers and other information
     */
    public void processEventReceived(String sender, byte[] receivedEventBytes) {
        int offset = 0;
        /** By passing the protocol Header */
        offset++;
        /** Retrieve the length of the event */
        byte[] lengthBytes = new byte[2];
        System.arraycopy(receivedEventBytes, offset, lengthBytes, 0, 2);
        offset += 2;
        int eventLength = (int) ByteUtilities.getShort(lengthBytes);

        /** Retrieve the event */
        byte[] eventBytes = new byte[eventLength];
        System
            .arraycopy(receivedEventBytes, offset, eventBytes, 0, eventLength);
        MEvent mEvent = new MEvent(eventBytes);
        long latency = System.currentTimeMillis() - mEvent.getTimeStamp();
        System.out.println("ClientProtocolHandler: Received event " + mEvent
            + " latency [" + latency + "] mSecs");
        /**
         * System.out.println("The received event is " + mEvent + " issued at
         * ->" + mEvent.getTimeStamp() + " and the latency is ->" + latency +
         * "mSecs");
         */

        if (gatherStats) {
            if (latencyTrack < latencyArray.length) {
                latencyArray[latencyTrack] = (short) latency;
            }
            if (latency < minLatency)
                minLatency = latency;
            if (latency > maxLatency)
                maxLatency = latency;

            latencyTrack++;
            if (latencyTrack == 1) {
                throughputStartTime = System.currentTimeMillis();
            }
            if (latencyTrack == latencyArray.length) {
                throughputEndTime = System.currentTimeMillis();
                printTestCaseDetails();
                if (loopThroughMessageSizes) {
                    testCases++;
                    clientNode.gatherStats(publishRates[publishRateTracker],
                                           matchRates[matchRateTracker],
                                           messageSizes[messageSizeTracker]);
                    messageSizeTracker++;
                    if (messageSizeTracker != messageSizes.length) {
                        initializeLatencyArray(numOfMessagesPerLoopIteration);
                        return;
                    }
                    publishRateTracker++;
                    if (publishRateTracker != publishRates.length) {
                        messageSizeTracker = 0;
                        initializeLatencyArray(numOfMessagesPerLoopIteration);
                        System.out
                            .println("Iterating to the next publish rate ->"
                                + publishRates[publishRateTracker]);
                        return;
                    }
                    matchRateTracker++;
                    if (matchRateTracker != matchRates.length) {
                        messageSizeTracker = 0;
                        numOfMessagesPerLoopIteration = matchRates[matchRateTracker];
                        initializeLatencyArray(numOfMessagesPerLoopIteration);
                        publishRateTracker = 0;
                        System.out
                            .println("Iterating to the next matching rate ->"
                                + matchRates[matchRateTracker]);
                        clientNode.initializeReportFile(reportFileName
                            + "WithMatchRate" + matchRates[matchRateTracker]);
                        return;
                    }
                    System.out.println("The Test Set involving (" + testCases
                        + ") is completed");
                }
                /** loop through message sizes */
            } /* latency track == latencyArray.length */
        } /* gather Stats */
    }

    /**
     * This method processes the packet that is received with the Protocol.EVENT
     * header. This method is reponsible for unmarshalling the event stream and
     * creating the event associated with the stream.
     * 
     * @param sender
     *            The server from which we received this event.
     * @param receivedEventBytes
     *            The byte stream representation of the event, including the
     *            protocol headers and other information
     */
    public void processPersistentEventReceived(String sender,
                                               byte[] receivedEventBytes) {
        int offset = 0;
        /** By passing the protocol Header */
        offset++;
        /** Retrieve the length of the event */
        byte[] lengthBytes = new byte[2];
        System.arraycopy(receivedEventBytes, offset, lengthBytes, 0, 2);
        offset += 2;
        int eventLength = (int) ByteUtilities.getShort(lengthBytes);

        /** Retrieve the event */
        byte[] eventBytes = new byte[eventLength];
        System
            .arraycopy(receivedEventBytes, offset, eventBytes, 0, eventLength);
        offset += eventLength;
        MEvent mEvent = new MEvent(eventBytes);

        long _epoch;
        byte[] epochBytes = new byte[8];
        System.arraycopy(receivedEventBytes, offset, epochBytes, 0, 8);
        _epoch = ByteUtilities.getLong(epochBytes);
        long latency = System.currentTimeMillis() - mEvent.getTimeStamp();

        /*
         * System.out.println("The received persistent event is " + mEvent +
         * "with an epoch of " + _epoch + " and the latency is ->" + latency +
         * "mSecs");
         */
        if (gatherStats) {
            if (latencyTrack < latencyArray.length) {
                latencyArray[latencyTrack] = (short) latency;
            }
            if (latency < minLatency)
                minLatency = latency;
            if (latency > maxLatency)
                maxLatency = latency;

            latencyTrack++;
            if (latencyTrack == 1) {
                throughputStartTime = System.currentTimeMillis();
            }
            if (latencyTrack == latencyArray.length) {
                throughputEndTime = System.currentTimeMillis();
                printTestCaseDetails();
                if (loopThroughMessageSizes) {
                    testCases++;
                    clientNode.gatherStats(publishRates[publishRateTracker],
                                           matchRates[matchRateTracker],
                                           messageSizes[messageSizeTracker]);
                    messageSizeTracker++;
                    if (messageSizeTracker != messageSizes.length) {
                        initializeLatencyArray(numOfMessagesPerLoopIteration);
                        return;
                    }
                    publishRateTracker++;
                    if (publishRateTracker != publishRates.length) {
                        messageSizeTracker = 0;
                        initializeLatencyArray(numOfMessagesPerLoopIteration);
                        System.out
                            .println("Iterating to the next publish rate ->"
                                + publishRates[publishRateTracker]);
                        return;
                    }
                    matchRateTracker++;
                    if (matchRateTracker != matchRates.length) {
                        messageSizeTracker = 0;
                        numOfMessagesPerLoopIteration = matchRates[matchRateTracker];
                        initializeLatencyArray(numOfMessagesPerLoopIteration);
                        publishRateTracker = 0;
                        System.out
                            .println("Iterating to the next matching rate ->"
                                + matchRates[matchRateTracker]);
                        clientNode.initializeReportFile(reportFileName
                            + "WithMatchRate" + matchRates[matchRateTracker]);
                        return;
                    }
                    System.out.println("The Test Set involving (" + testCases
                        + ") is completed");
                }
                /** loop through message sizes */
            } /* latency track == latencyArray.length */
        } /* gather Stats */

    }

    /**
     * Process the response that you have received from the server pertaining to
     * the address request that was issued. The responses could be one of the
     * following.
     * <ul>
     * <li>The server to which this request was issued, hasn't been assigned a
     * logical address by the system.
     * <li>The server has possibly exhausted the number of open connections it
     * can maintain for the clients that are connected to it.
     * <li>The request was successfully executed and completed. The response
     * contains the destinationID of the client, the server destination.
     * </ul>
     * 
     * @param sender -
     *            The server from which this response was received.
     * @param addressResponse -
     *            The byte stream representation of the address response
     */
    public void processNodeAddressResponse(String sender, byte[] addressResponse) {
        if (addressResponse.length == 1) {
            System.out.println("This server node (" + sender
                + ") hasn't been assigned an address yet ");
        }
        if (addressResponse.length == 2) {
            System.out.println("The system has exceeded its capacity to handle"
                + " clients, please try at a different node ");
        }

        int offset = 0;
        offset++;
        /** Skipping the header */
        byte[] destinationBytes = new byte[4];
        System.arraycopy(addressResponse, offset, destinationBytes, 0, 4);
        destinations = ByteUtilities.getInt(destinationBytes);
        offset += 4;
        byte[] serverDestBytes = new byte[addressResponse.length - offset];
        System.arraycopy(addressResponse, offset, serverDestBytes, 0,
                         serverDestBytes.length);
        serverDestination = new Destinations(serverDestBytes);
        System.out
            .println("This node has been assigned the address "
                + ByteUtilities.printInt(destinations)
                + " and it is connected to the following node "
                + serverDestination);

    }

    /**
     * This method is reponsible for issuing a node address request to the
     * server in question.
     * 
     * @param server
     *            The server to which this request is being issued.
     */
    public void nodeAddressRequest(String server) {
        byte[] data = new byte[1];
        data[0] = Protocol.NAP_CLIENT_ADD;
        sendTo(server, data);
    }

    /**
     * Propagates the interest in a subscription to the server in question. This
     * method cannot be issued unitil the client has been assigned a logical
     * address by the system.
     * 
     * @param server
     *            The server to which this client is attached.
     * @param subscription
     *            The subscription predicate for the client
     */
    public void profilePropagationRequest(String server, String subscription) {
        Predicate predicate = new Predicate(subscription, server, destinations);
        byte[] predicateBytes = predicate.getBytes();
        int predicateLength = predicateBytes.length;
        byte[] propagateSubscription = new byte[1 /* protocol header */+ 4 /*
                                                                             * predicate
                                                                             * length
                                                                             */+ predicateLength];
        int offset = 0;
        propagateSubscription[offset++] = Protocol.PPP_CLIENT_ADD_PREDICATE;
        // propagateSubscription[offset++] = (byte) predicateLength;
        System.arraycopy(ByteUtilities.getBytes(predicateLength), 0,
                         propagateSubscription, offset, 4);
        offset += 4;
        System.arraycopy(predicateBytes, 0, propagateSubscription, offset,
                         predicateLength);
        sendTo(server, propagateSubscription);
    }

    /**
     * This method is responsible for publishing an event to the system.
     * 
     * @param server
     *            The server to which this request is being issued.
     * @param event
     *            The event that is being published
     * @param minSize
     *            Minimum Size of the event being published.
     */
    public void eventPropagationRequest(String server, String event, int minSize) {
        MEvent mEvent = new MEvent(event, minSize);
        byte[] eventBytes = mEvent.getBytes();
        int eventLength = eventBytes.length;
        byte[] propagateEvent = new byte[1 /* protocol header */+ 2 /*
                                                                     * event
                                                                     * length
                                                                     */+ eventLength];
        int offset = 0;
        propagateEvent[offset++] = Protocol.CLIENT_EVENT;

        /** Marshalling the event length in two bytes */
        System.arraycopy(ByteUtilities.getBytes((short) eventLength), 0,
                         propagateEvent, offset, 2);
        offset += 2;
        // propagateEvent[offset++] = (byte) eventLength;

        System.arraycopy(eventBytes, 0, propagateEvent, offset, eventLength);
        sendTo(server, propagateEvent);
    }

    /**
     * This method is responsible for publishing a persistent event to the
     * system.
     * 
     * @param server
     *            The server to which this request is being issued.
     * @param event
     *            The event that is being published
     * @param minSize
     *            Minimum Size of the event being published.
     */
    public void persistentEventPropagationRequest(String server, String event,
                                                  int minSize) {
        MEvent mEvent = new MEvent(event, minSize);
        byte[] eventBytes = mEvent.getBytes();
        int eventLength = eventBytes.length;
        byte[] propagateEvent = new byte[1 /* protocol header */+ 2 /*
                                                                     * event
                                                                     * length
                                                                     */+ eventLength];
        int offset = 0;
        propagateEvent[offset++] = Protocol.CLIENT_PERSISTENT_EVENT;

        /** Marshalling the event length in two bytes */
        System.arraycopy(ByteUtilities.getBytes((short) eventLength), 0,
                         propagateEvent, offset, 2);
        offset += 2;
        // propagateEvent[offset++] = (byte) eventLength;

        System.arraycopy(eventBytes, 0, propagateEvent, offset, eventLength);
        sendTo(server, propagateEvent);
    }

    public void loadCommunicationsOfType(Properties props, String commType)
                                                                           throws TransportException {
        transportHandler.loadCommunicationsOfType(props, commType);
    }

    public String setupLink(Properties props, String linkType)
                                                              throws TransportException {
        return transportHandler.setupLink(props, linkType);
    }

    /**
     * For communicating with a server node which hasn't been assigned a logical
     * address so far
     * 
     * @param node
     *            The string identifier for the server node.
     * @param data
     *            The byte stream containing the information.
     */
    public void sendTo(String node, byte[] data) {
        try {
            transportHandler.sendData(data, node);
        } catch (TransportException transEx) {
            System.out.println(transEx);
        }
    }

    /**
     * This method send information over a specified hop. The information
     * maintained in the connectivity graph is just the level, and the node
     * address at that level
     */
    public void sendToNode(Gateway nodeAddress, byte[] data) {
        try {
            transportHandler.sendData(data, nodeAddress);
        } catch (TransportException transEx) {
            System.out.println(transEx);
        }
    }

    public void run() {
    }

    public static void main(String[] args) {
        ClientProtocolHandler protocolHandler = new ClientProtocolHandler();
        protocolHandler.start();
        BufferedReader r = new BufferedReader(new InputStreamReader(System.in));
        String cmd;
        StringTokenizer tok;
        System.out.println("Program Exiting Now");
    }

}
